package store

import (
	"sync"
	"sync/atomic"
	"time"
)

// SyncRunner call log sync
type SyncRunner struct {
	log    *MMapLog
	synced *uint64

	close   bool
	hasData chan struct{}
	syncing *uint32

	lock *sync.Mutex
	cond *sync.Cond
}

const sleepWaitGroup = 10 * time.Nanosecond

func NewSyncRunner(log *MMapLog, minSync time.Duration) *SyncRunner {
	lock := new(sync.Mutex)
	syncRunner := &SyncRunner{
		log:     log,
		synced:  new(uint64),
		close:   false,
		hasData: make(chan struct{}, 1),
		syncing: new(uint32),
		lock:    lock,
		cond:    sync.NewCond(lock),
	}
	go syncRunner.Run(minSync)
	return syncRunner
}

func (s *SyncRunner) Run(minSync time.Duration) {
	ticker := time.Tick(minSync)
	for true {
		if s.close {
			break
		}
		select {
		case <-ticker:
			break
		case <-s.hasData:
			break
		}
		tail := atomic.LoadUint64(s.synced)
		time.Sleep(sleepWaitGroup)
		for tail < s.log.Last() {
			tail = s.log.Last()
			time.Sleep(sleepWaitGroup)
		}
		if tail == atomic.LoadUint64(s.synced) {
			continue
		}
		err := s.log.Sync()
		if err != nil {
			panic(err)
		}
		atomic.StoreUint64(s.synced, tail)
		s.Broadcast()
	}
}

func (s *SyncRunner) WeakUp() {
	if atomic.CompareAndSwapUint32(s.syncing, 0, 1) {
		s.hasData <- struct{}{}
	}
}

func (s *SyncRunner) Broadcast() {
	s.lock.Lock()
	s.cond.Broadcast()
	s.lock.Unlock()
}

func (s *SyncRunner) WaitDone(index uint64) {
	s.cond.L.Lock()
	// break on seq <= synced
	for index > atomic.LoadUint64(s.synced) {
		s.cond.Wait()
	}
	s.cond.L.Unlock()
}
